package pers.cz.javacvdemo.oss.service;

import com.lmax.disruptor.*;
import com.lmax.disruptor.dsl.ProducerType;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Component;
import pers.cz.javacvdemo.oss.model.QueueDataWrapper;

import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.Executors;

/**
 * @version 1.0
 * @description:
 * @author: ChengZhi 0336
 * @create: 2025-08-14 14:33
 **/
@Slf4j
@Component
public class RingBufferWorkerFactory {

	private final Map<Integer, BucketEventProducer> producers = new ConcurrentHashMap<>();

	private RingBuffer<QueueDataWrapper> ringBuffer;

	public void initRingBuffer(BucketEventHandler[] messageConsumers) {
		int mBufferSize = 2048;
		this.ringBuffer = RingBuffer.create(ProducerType.MULTI,
				QueueDataWrapper::new,
				mBufferSize,
				new BlockingWaitStrategy());
		// 屏障: 防止消费者读取到未写入完成的数据
		SequenceBarrier sequenceBarrier = this.ringBuffer.newBarrier();
		WorkerPool<QueueDataWrapper> workerPool = new WorkerPool<>(
				this.ringBuffer,
				sequenceBarrier,
				new EventExceptionHandler(),
				messageConsumers);
		this.ringBuffer.addGatingSequences(workerPool.getWorkerSequences());
		int processorsCount = Runtime.getRuntime().availableProcessors();
		log.info("进程数 -> {}", processorsCount);
		workerPool.start(Executors.newFixedThreadPool(processorsCount));
	}

	public BucketEventProducer getMessageProducer() {
		return new BucketEventProducer(this.ringBuffer);
	}

	/**
	 * 异常静态类
	 */
	@Slf4j
	static class EventExceptionHandler implements ExceptionHandler<QueueDataWrapper> {

		@Override
		public void handleEventException(Throwable ex, long sequence, QueueDataWrapper event) {
			log.error("handleEventException -> ex:{}  sequence:{} event:{}", ex.getMessage(), sequence, event.getClass().toString());
			ex.printStackTrace();
		}

		@Override
		public void handleOnStartException(Throwable ex) {
			log.error("handleOnStartException -> ex:{}", ex.getMessage());
			ex.printStackTrace();
		}

		@Override
		public void handleOnShutdownException(Throwable ex) {
			log.error("handleOnShutdownException -> ex:{} ", ex.getMessage());
			ex.printStackTrace();
		}
	}


}
